package com.zlm.app;

import com.zlm.bean.ItemViewCount;
import com.zlm.bean.UserBehavior;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Properties;

/**
 * Author: Harbour
 * Date: 2021-05-13 16:42
 * Desc: 实时热门商品统计
 */
public class HotItemsApp {
    public static void main(String[] args) throws Exception {
        // step 1 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); // 引入时间语义

        // step 2 读取数据源
//        DataStreamSource<String> inputStream =
//                env.readTextFile("D:\\JavaProject\\real-time-data-warehouse\\hot-item\\src\\main\\resources\\UserBehavior.csv");

        // kafka 配置项
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "Hadoop201:9092");
        properties.setProperty("group.id", "consumer-group");
        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("auto.offset.reset", "latest");

        // 读取kafka数据源
        DataStream<String> inputStream = env.addSource(new FlinkKafkaConsumer<>("hot-item", new SimpleStringSchema(), properties));


        // step 3 将数据源转换成 指定的java bean得到 data stream
        DataStream<UserBehavior> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new UserBehavior(
                    new Long(fields[0]),
                    new Long(fields[1]),
                    new Integer(fields[2]),
                    fields[3],
                    new Long(fields[4])
            );
        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
            // 设定如何提取数据中的Event time
            @Override
            public long extractAscendingTimestamp(UserBehavior element) {
                // 数据中提供的是秒，而此函数所用的时间戳需要毫秒
                return element.getTimestamp() * 1000L;
            }
        });

        // step 4 分组开窗聚合， 得到每个窗口内各个商铺的count值
        DataStream<ItemViewCount> aggregateStream = dataStream.filter(data -> "pv".equalsIgnoreCase(data.getBehavior()))
                .keyBy("itemId") // 聚合
                .timeWindow(Time.hours(1), Time.minutes(5)) // 开滑动窗口（窗口大小，滑动大小）
                .aggregate( // 累加
                        new MyUserBehaviorIncrementAggregationFunctions(),
                        new MyUserBehaviorFullWindowFunctions());

        aggregateStream.keyBy("windowEnd")
                .process(new MyTopNHotItemsProcessFunctions(5))
                .print();

        env.execute("Hot item analysis");
    }

    /**
     * 时间窗口增量累加函数类
     */
    private static class MyUserBehaviorIncrementAggregationFunctions implements AggregateFunction<UserBehavior, Long, Long> {

        @Override
        public Long createAccumulator() {
            // 创建累加器的初始值
            return 0L;
        }

        @Override
        public Long add(UserBehavior value, Long accumulator) {
            return accumulator + 1;
        }

        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return a + b;
        }
    }

    private static class MyUserBehaviorFullWindowFunctions implements WindowFunction<Long, ItemViewCount, Tuple, TimeWindow> {

        @Override
        public void apply(Tuple tuple, TimeWindow window, Iterable<Long> input, Collector<ItemViewCount> out) throws Exception {
            Long itemId = tuple.getField(0);
            Long windowEnd = window.getEnd();
            Long count = input.iterator().next();
            out.collect(new ItemViewCount(itemId, windowEnd, count));
        }
    }

    private static class MyTopNHotItemsProcessFunctions extends KeyedProcessFunction<Tuple, ItemViewCount, String> {

        private final Integer topSize;
        ListState<ItemViewCount> itemViewCountListState;  // 保存所有itemViewCount的状态列表

        public MyTopNHotItemsProcessFunctions(int topSize) {
            this.topSize = topSize;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            // 富函数用来获取运行时上下文环境
            itemViewCountListState = getRuntimeContext().getListState(new ListStateDescriptor<>("itemViewCount-List", ItemViewCount.class));
        }

        @Override
        public void processElement(ItemViewCount value, Context ctx, Collector<String> out) throws Exception {
            itemViewCountListState.add(value); // 收集所有的itemViewCount
            ctx.timerService().registerProcessingTimeTimer(value.getWindowEnd() + 1); // 注册定时器
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            // 触发定时器，将当前所有数据进行排序
            ArrayList<ItemViewCount> itemViewCountList = Lists.newArrayList(itemViewCountListState.get().iterator());
            itemViewCountList.sort((e1, e2) -> (Long.compare(e1.getCount(), e2.getCount()) * -1));

            // 组装成String 准备输出
            StringBuilder output = new StringBuilder();
            output.append("=====================");
            output.append("窗口结束时间为：").append(new Timestamp(timestamp - 1)).append("\n");

            for (int i = 0; i < Math.min(topSize, itemViewCountList.size()); i++) {
                ItemViewCount itemViewCount = itemViewCountList.get(i);
                output.append("Top").append((i + 1)).append("是：").append(itemViewCount).append("\n");
            }

            Thread.sleep(1000L);

            out.collect(output.toString());
        }
    }
}
